package com.pushtime.service.impl.search;

import com.pushtime.service.search.SimpleHdfsUploadService;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.*;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.multipart.MultipartFile;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

@Service
@Transactional
public class SimpleHdfsUploadServiceImpl implements SimpleHdfsUploadService {

    @Value("${application.webhdfs.url}")
    private String url;
    @Value("${application.webhdfs.path}")
    private String path;

    private final RestTemplate restTemplate;

    public SimpleHdfsUploadServiceImpl(RestTemplate restTemplate) {
        this.restTemplate = restTemplate;
    }

    @Override
    public Map<String, String> upload(MultipartFile file) {
        Map<String, String> map = new HashMap<>();
        if (!file.isEmpty() && file.getOriginalFilename() != null) {
            String[] split = file.getOriginalFilename().split("\\.");
            String fileExt = "." + split[split.length - 1];
            // 格式化文件信息
            SimpleDateFormat dateFormat = new SimpleDateFormat("/yyyy/MM/dd/");
            Date date = Calendar.getInstance().getTime();
            String dir = dateFormat.format(date);
            long timestamp = date.toInstant().toEpochMilli();

            try {
                // 拼装请求头信息
                HttpHeaders headers = new HttpHeaders();
                HttpEntity<String> requestEntity = new HttpEntity<>("", headers);
                // 拼装参数信息
                Map<String, Object> params = new HashMap<>();
                params.put("op", "CREATE");
                byte[] bytes = file.getBytes();
                ResponseEntity<Object> exchangeEntity = restTemplate
                    .exchange(url + path + dir + timestamp + fileExt + "?op={op}",
                        HttpMethod.PUT, requestEntity, Object.class, params);

                // 如果获得 302 跳转，则进行下面的操作
                if (exchangeEntity.getStatusCode().equals(HttpStatus.TEMPORARY_REDIRECT)) {
                    URI location = exchangeEntity.getHeaders().getLocation();
                    if (location != null) {
                        HttpEntity<byte[]> sendEntity = new HttpEntity<>(bytes, new HttpHeaders());
                        ResponseEntity<Object> exchangeResponse = restTemplate.exchange(location,
                            HttpMethod.PUT,
                            sendEntity, Object.class);
                        map.put("state", "success");
                        map.put("src", url + path + dir + timestamp + fileExt + "?op=OPEN");
                        map.put("path", dir + timestamp + fileExt + "?op=OPEN");
                        return map;
                    }
                } else {
                    // 返回状态码错误
                    map.put("msg", "hdfs connect error!");
                }

            } catch (Exception e) {
                map.put("msg", e.getMessage());
            }

        }
        map.put("state", "error");
        return map;
    }
}
